package DianShang_2024.ds_server.clean

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.{current_timestamp, date_format, lit, to_timestamp}

object extract02 {
  def main(args: Array[String]): Unit = {
    /*
          2、抽取ods库中product_info表中昨天的分区（任务一生成的分区）数据，并结合dim_product_info最新分区现有的数据，根据product_id合并数据到dwd库中
          dim_product_info的分区表（合并是指对dwd层数据进行插入或修改，需修改的数据以product_id为合并字段，根据modified_time排序取最新的一条），分区字段为
          etl_date且值与ods库的相对应表该值相等，并添加dwd_insert_user、dwd_insert_time、dwd_modify_user、dwd_modify_time四列,其中
          dwd_insert_user、dwd_modify_user均填写“user1”。若该条记录第一次进入数仓dwd层则dwd_insert_time、dwd_modify_time均存当前操作时间，
          并进行数据类型转换。若该数据在进入dwd层时发生了合并修改，则dwd_insert_time时间不变，dwd_modify_time存当前操作时间，其余列存最新的值。
          使用hive cli执行show partitions dwd.dim_product_info命令；
     */
    //  准备 sparksql的环境
    val spark=SparkSession.builder()
      .master("local[*]")
      .appName("数据清洗第二题")
      .enableHiveSupport()
      .getOrCreate()

    spark.sql("use dwd_server")

    /*
          处理逻辑:1.首先对ods_server.product_info01表进行添加字段操作，并且创建临时表
                      2.对两个表的数据进行完全合并操作
                      3.对合并后的表进行开窗函数，分区排序得到最新的数据
     */

//  对ods_server里面的表进行添加字段
    spark.sql(
      """
        |select * from ods_server.product_info01 where etl_date='20240311'
        |""".stripMargin)
      .withColumn("dwd_insert_user",lit("user1"))
      .withColumn(
        "dwd_insert_time",
        to_timestamp(date_format(current_timestamp(),"yyyy-MM-dd HH:mm:ss"),"yyyy-MM-dd HH:mm:ss")
      )
      .withColumn("dwd_modify_user",lit("user1"))
      .withColumn(
        "dwd_modify_time",
        to_timestamp(date_format(current_timestamp(),"yyyy-MM-dd HH:mm:ss"))
      )
      .createOrReplaceTempView("temp_table")

    //  首先对两表进行合并，然后进行开窗选择数据
    spark.sql(
      """
        |select
        |product_id ,
        |product_core ,
        |product_name ,
        |bar_code ,
        |brand_id,
        |one_category_id ,
        |two_category_id ,
        |three_category_id ,
        |supplier_id ,
        |price ,
        |average_cost ,
        |publish_status ,
        |audit_status ,
        |weight ,
        |length ,
        |height,
        |width ,
        |color_type ,
        |production_date ,
        |shelf_life ,
        |descript ,
        |indate ,
        |modified_time ,
        |row_number() over(partition by product_id order by modified_time desc) as row_number,
        |dwd_insert_user,
        |min(dwd_insert_time) over(partition by product_id) as dwd_insert_time,
        |dwd_modify_user,
        |dwd_modify_time
        |from(
        |select * from dwd_server.dim_product_info where etl_date='20240311'
        |union all
        |select * from temp_table
        |) as t1
        |""".stripMargin).createOrReplaceTempView("all_datas")

    //  最后根据条件插入数据
    spark.sql(
      """
        |select
        |product_id ,
        |product_core ,
        |product_name ,
        |bar_code ,
        |brand_id,
        |one_category_id ,
        |two_category_id ,
        |three_category_id ,
        |supplier_id ,
        |price ,
        |average_cost ,
        |publish_status ,
        |audit_status ,
        |weight ,
        |length ,
        |height,
        |width ,
        |color_type ,
        |production_date ,
        |shelf_life ,
        |descript ,
        |indate ,
        |modified_time ,
        |dwd_insert_user,
        |dwd_insert_time,
        |dwd_modify_user,
        |dwd_modify_time
        |from all_datas
        |where row_number=1
        |""".stripMargin)





    //  关闭sparksql的环境
    spark.close()
  }

}
